package org.lds.simple.disruptor;

import lombok.Data;
import org.lds.simple.disruptor.consumer.exception.ConsumeExceptionHandler;
import org.lds.simple.disruptor.consumer.EventConsumer;
import org.lds.simple.disruptor.consumer.EventConsumerProcessor;
import org.lds.simple.disruptor.event.Event;
import org.lds.simple.disruptor.event.EventFactory;
import org.lds.simple.disruptor.producer.sequencer.MultiProducerSequencer;
import org.lds.simple.disruptor.producer.ProducerType;
import org.lds.simple.disruptor.producer.sequencer.Sequencer;
import org.lds.simple.disruptor.producer.sequencer.SingleProducerSequencer;
import org.lds.simple.disruptor.strategy.WaitStrategy;

@Data
public class Disruptor<T extends Event> implements LifeCycle {
    private RingBuffer<T> ringBuffer;

    private EventConsumerProcessor eventConsumerProcessor;

    public Disruptor(int bufferSize, EventFactory eventFactory, ProducerType producerType, WaitStrategy waitStrategy) {
        this(bufferSize, eventFactory, producerType, waitStrategy, null);
    }

    public Disruptor(int bufferSize, EventFactory eventFactory, ProducerType producerType, WaitStrategy waitStrategy, ConsumeExceptionHandler consumeExceptionHandler) {
        Sequencer sequencer = buildSequencer(producerType);
        this.ringBuffer = new RingBuffer<>(bufferSize, eventFactory, sequencer);
        this.eventConsumerProcessor = new EventConsumerProcessor(sequencer, ringBuffer, waitStrategy, consumeExceptionHandler);
    }

    private Sequencer buildSequencer(ProducerType producerType) {
        if (producerType == ProducerType.SINGLE) return new SingleProducerSequencer();
        return new MultiProducerSequencer();
    }

    public void addConsumer(Class<? extends EventConsumer> consumerClass) {
        eventConsumerProcessor.register(consumerClass);
    }

    @Override
    public void start() {
        ringBuffer.start();
        eventConsumerProcessor.start();
    }

    @Override
    public void shutdown() {
        // 不再生产消息
        ringBuffer.shutdown();

        // 已经生产的消息一定要消费完
        // 所有消息消费完毕后线程退出
        eventConsumerProcessor.shutdown();
    }
}
